package Yanxin.atguigu.yx.app.App_03_DwdApp.Dwd_01_Traffic;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import Yanxin.atguigu.yx.app.APP_01_BaseApp.BaseApp;
import Yanxin.atguigu.yx.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.PatternTimeoutFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.OutputTag;
import Yanxin.atguigu.yx.common.Constant;

import java.time.Duration;
import java.util.List;
import java.util.Map;


public class Dwd_02_Traffic_UserJumpDetail extends BaseApp {
    /**
     * 需求说明：跳出行为指的是用户在一个网页停留后，没有再进入该页面的子页面（last_page 为 null）
     *      1、如果一个用户短时间内反复刷新一个页面，也算是跳出
     *      2、一个用户在一个页面上停留时间超过5s都没有进入该页面的子页面也算跳出
     *      3、补充一点，如果CEP出现匹配不满足条件，则属于匹配错误，不会有超时的情况出现
     *      例如这个就属于匹配错误，第二条数据已经不满足cep的过滤规则，整个都会被当成匹配错误，ts为16000秒这条数据不会被当成超时
     *                 "{"common":{"mid":"102"},"page":{"page_id":"home"},"ts":16000}",
     *                 "{"common":{"mid":"102"},"page":{"page_id":"good_list","last_page_id":"home"},"ts":18000} "
     */
    public static void main(String[] args) {
        Dwd_02_Traffic_UserJumpDetail dwd_02_traffic_userJumpDetail = new Dwd_02_Traffic_UserJumpDetail();

        dwd_02_traffic_userJumpDetail.init(
                3003,
                1,
                "Dwd_02_Traffic_UserJumpDetail",
                Constant.TOPIC_DWD_TRAFFIC_PAGE //数据源于dwd_traffic_page这个topic，可以从这里dwd_traffic_page
        );

    }

    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {

       SingleOutputStreamOperator<JSONObject> etl = etl(stream);

        //从kafka消费的数据按照mid进行分组，并且设置水位线
        KeyedStream<JSONObject, String> keyedStream = etl
//                .map(stringJson -> JSON.parseObject(stringJson))
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
//                                .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(0)) //测试数据用
                                .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((obj, ts) -> obj.getLong("ts") )
                )
                .keyBy(obj -> obj.getJSONObject("common").getString("mid"));

        //1.定义规则
        Pattern<JSONObject, JSONObject> pattern = Pattern
                .<JSONObject>begin("page_enter_first")
                .where(new SimpleCondition<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject value) throws Exception {
                        /**
                         * 数据形态
                         * {
                         *   "common": {
                         *     "sc": "4",
                         *     "ar": "12",
                         *     "uid": "97",
                         *     "os": "Android 11.0",
                         *     "ch": "xiaomi",
                         *     "is_new": "0",
                         *     "md": "Xiaomi 10 Pro ",
                         *     "mid": "mid_495",
                         *     "vc": "v2.1.134",
                         *     "ba": "Xiaomi",
                         *     "sid": "c0426024-72bb-4665-89b0-c7b3ef154eab"
                         *   },
                         *   "page": {
                         *     "page_id": "course_list",
                         *     "item": "java",
                         *     "during_time": 7060,
                         *     "item_type": "keyword"
                         *   },
                         *   "ts": 1661926536129
                         * }
                         *
                         *
                         */
                        //补充一点，有些数据甚至都没有last_page_id这个kv
                        String last_page_id = value.getJSONObject("page").getString("last_page_id");


                        return last_page_id == null || last_page_id.length() == 0;
                    }
                })

                //todo 如果某个用户第一次第二次都进入同一个页面，这两次行为都是跳出行为
                .next("page_enter_second")
                .where(new SimpleCondition<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject value) throws Exception {
                        String last_page_id = value.getJSONObject("page").getString("last_page_id");

//                        System.out.println("-------next---------");
//                        System.out.println(last_page_id);
//                        System.out.println("-------next--------");

                        //last_page有可能为null或者""空字符串
                        return last_page_id == null || last_page_id.length() == 0;
                    }
                })

                //todo 短时间内刷新同一个网页也是跳出行为判定方法
                .within(Time.seconds(5));


        //2.将规则作用到流上
        PatternStream<JSONObject> patternStream = CEP.pattern(keyedStream, pattern);


        //3.处理最后一种情况，用户在同一个页面停留时间过长，并且将最终结果输出
        // todo 在pattern的匹配规则下，超过了within的时间，那么就是超时的数据，
        //  这类数据属于用户看一个页面很久不操作，大概率是忘记关掉页面，也属于是跳出
        SingleOutputStreamOperator<JSONObject> result = patternStream
                .select(
                        //分两种情况：
                        //1.实例化对象是class，则大括号为初始化该对象
                        //2.通过匿名内部类实现接口方法，实例化接口
                        new OutputTag<JSONObject>("timeout") {},

                        //将超时的数据拿出来，放到侧输出流中
                        new PatternTimeoutFunction<JSONObject, JSONObject>() {
                            @Override
                            public JSONObject timeout(Map<String, List<JSONObject>> pattern,
                                                      long timeoutTimestamp) throws Exception {
//                                System.out.println(pattern.get("page_enter_first").get(0));
                                return pattern.get("page_enter_first").get(0);
                            }
                        },

                        //对外展示数据
                        new PatternSelectFunction<JSONObject, JSONObject>() {
                            @Override
                            public JSONObject select(Map<String, List<JSONObject>> pattern) throws Exception {
                                return pattern.get("page_enter_first").get(0);
                            }
                        }

                );

        //4.将数据发送到kafka中，并且放到一个新的topic中
//        result.getSideOutput(new OutputTag<JSONObject>("timeout"){}).print("超时");
//        result.print("主流cep匹配短时间刷新");


        result.getSideOutput(new OutputTag<JSONObject>("timeout") {}).union(result)
                .map(jsonObject -> jsonObject.toJSONString())
                .addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_UJ_DETAIL));

    }

    private SingleOutputStreamOperator<JSONObject> etl(DataStreamSource<String> stream) {

        //只要json数据
        return stream
                .filter(jsonString -> {
                    if (jsonString.equals("")) {
                        System.out.println(jsonString + "jsonString是一个空字符串，JSON.parseObject(jsonString)会将其转化成null");
                        return false;
                    }

                    try {
                        JSON.parseObject(jsonString);
                    } catch (Exception e) {
                        System.out.println("json格式有误，请检查：" + jsonString);
                        e.printStackTrace();
                        return false;
                    }
                    return true;
                })
//                .map(json->{return JSON.parseObject(json);})
                .map(JSON::parseObject);
//                .print();


    }
}
